package net.bw.realtime.tmall.dws.function;

import net.bw.realtime.tmall.dwd.bean.TransactionEventBean;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/*
 * @ Author：liuyawei
 * @ Date：2025-05-27
 */
public class TmallOrderCountFunction implements WindowFunction<TransactionEventBean, String, String, TimeWindow> {

    private FastDateFormat fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

    @Override
    public void apply(String s, TimeWindow window, Iterable<TransactionEventBean> input, Collector<String> out) throws Exception {

        //   1.获取窗口开始和结束时间
        String windowStartTime = fastDateFormat.format(window.getStart());
        String windowEndTime = fastDateFormat.format(window.getEnd());

        //   2.获取订单数量
        long order_count = 1L;

        for (TransactionEventBean transactionEventBean : input) {
            order_count += 1L;
        }

        String output = windowStartTime + "," + windowEndTime + "," + order_count + "," + System.currentTimeMillis();

        out.collect(output);

    }
}
